package com.etlmaster.executor.processors.File;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Reader;
import java.io.UnsupportedEncodingException;
import java.io.Writer;

import javax.script.Bindings;
import javax.script.Compilable;
import javax.script.CompiledScript;
import javax.script.ScriptEngine;
import javax.script.ScriptEngineManager;
import javax.script.ScriptException;

import org.apache.commons.io.IOUtils;
import org.apache.commons.vfs2.FileObject;
import org.apache.commons.vfs2.FileSystemException;
import org.apache.commons.vfs2.FileType;
import org.apache.commons.vfs2.Selectors;

import com.etlmaster.executor.utils.CompressUtil;
import com.etlmaster.executor.utils.DpVfsUtils;
import com.etlmaster.executor.utils.LineReader;
import com.etlmaster.executor.utils.LogWriter;
import com.etlmaster.executor.utils.StringUtils;

/**
 * 文件传输器
 * 
 * @author Wugf
 * @date 2016-03-08
 */
public class FileTransfer {
	FileObject sourceFile = null;
	protected FileObject targetTempFile = null;
	protected FileObject targetFile = null;
	
	FileObject source=null;
	FileObject target=null;

	InputStream in = null;
	protected OutputStream out = null;

	Reader reader = null;
	protected Writer writer = null;

	protected long sourceSize=0;
	LineReader bufferReader = null;
	protected BufferedWriter bufferWriter = null;
	private static  final String splitBySize="size";
	private static  final String splitByRowNum="rowNum";

	private CompiledScript compiledScript;
	private Bindings bindings;

	private long timestamp=System.currentTimeMillis();
	protected FileTransInfo fileTransInfo;
	protected long transSize = 0;

	protected int counter=1;
	public FileTransfer(FileTransInfo filetransInfo) {
		this.fileTransInfo = filetransInfo;
	}

	private void setUpInputOutPutStream() throws FileSystemException {
		
		
		
	}

	private void setUpSouceObjects(String sourceVfsURI) throws FileSystemException, UnsupportedEncodingException {
		sourceFile = DpVfsUtils.getInstance().resolveFile(
				sourceVfsURI);
		in = sourceFile.getContent().getInputStream();
		LogWriter.addLog("INFO", "源字符集为{}",
				fileTransInfo.getSourceCharSet());
		reader = new InputStreamReader(in, fileTransInfo.getSourceCharSet());
	}
	public void setUpTargetObjects(String targetVfsURI) throws FileSystemException, UnsupportedEncodingException{
		targetTempFile = DpVfsUtils.getInstance().resolveFile(
				targetVfsURI+ ".tmp");
		targetFile = DpVfsUtils.getInstance().resolveFile(
				targetVfsURI);
		if (fileTransInfo.isAppend() && targetFile.exists()
				&& targetFile.getType() == FileType.FILE) {// 如果追加写，就先把存在的目标文件先改成.tmp
			sourceSize=targetFile.getContent().getSize();
			targetFile.moveTo(targetTempFile);
		}
		out = targetTempFile.getContent().getOutputStream(
				fileTransInfo.isAppend());
		LogWriter.addLog("INFO", "目标字符集为{}",
				fileTransInfo.getTargetCharSet());
		writer = new OutputStreamWriter(out,
				fileTransInfo.getTargetCharSet());
	}

	public boolean process() {
		try {
			setUpSouceObjects(fileTransInfo.getSourceVfsURI());
			if (sourceFile.exists() && sourceFile.getType() == FileType.FILE) {
				fileTransInfo.setSourceFileSize(sourceFile.getContent().getSize());
				setUpTargetObjects(fileTransInfo.getTargetVfsURI());
				//如果是压缩，则调用解压方法，如果是保留原压缩文件，则继续执行copyFile，否则直接返回
				if (fileTransInfo.isDeCompress() && !fileTransInfo.isKeepBack()) {
					copyCompressFile();
					return true;
				} else {
					copyFile();
				}

				closeInStream();
				closeOutStream();
			} else {
				LogWriter.addLog("ERROR", "源文件{}不存在或不是文件，无法写入",
						fileTransInfo.getSourceVfsURI());

				return false;
			}
		} catch (Exception e) {
			LogWriter.addLog("ERROR", "文件{}写入到{}失败,{}",
					fileTransInfo.getSourceVfsURI(),
					fileTransInfo.getTargetVfsURI(), e.getMessage());
			return false;

		} finally {
			IOUtils.closeQuietly(bufferWriter);
			IOUtils.closeQuietly(reader);
			IOUtils.closeQuietly(writer);
			IOUtils.closeQuietly(in);
			IOUtils.closeQuietly(out);
			DpVfsUtils.getInstance().closeFileObject(sourceFile);
			DpVfsUtils.getInstance().closeFileObject(targetTempFile);
			DpVfsUtils.getInstance().closeFileObject(targetFile);
		}
		return true;
	}

	private void copyFile() throws IOException, ScriptException {
		long startTime = System.currentTimeMillis();
		setUpInputOutPutStream();
		if (fileTransInfo.isRowRead()) {
			
			copyFileByLine();
		} else {
			copyFileBySize();
		}
		targetTempFile.moveTo(targetFile);
		fileTransInfo.setTransFileSize(transSize);
		fileTransInfo.setTransDura(System.currentTimeMillis() - startTime);
		LogWriter.addLog("INFO",
				"文件{}写入到{}成功,文件原始大小{}，传输大小{}，传输行数{}，传输耗时{} s，传输速度{} MB/S",
				fileTransInfo.getSourceVfsURI(),
				fileTransInfo.getTargetVfsURI(),
				fileTransInfo.getSourceFileSize() + "",
				fileTransInfo.getTransFileSize() + "",
				fileTransInfo.getRowNum() + "", fileTransInfo.getTransDura()
						/ 1000.0 + "", fileTransInfo.getTransFileSize()
						/ 1000.0 / fileTransInfo.getTransDura() + "");

	}
	private void copyCompressFile(){
		try{
			LogWriter.addLog("INFO", "使用解压");
			
			if (targetFile.exists() && targetFile.getType() == FileType.FOLDER)
			{
				target = DpVfsUtils.getInstance().resolveFile(targetFile.getName().toString());
			}else{
				target = DpVfsUtils.getInstance().resolveFile(targetFile.getName().getParent().toString());
			}
			LogWriter.addLog("DEBUG", "源文件URL：{}", sourceFile.toString());
			String unCompressUri = CompressUtil.getInstance().unCompressUri(sourceFile.toString());
			LogWriter.addLog("DEBUG", "解压基线压缩文件URL：{}", unCompressUri);
			source = DpVfsUtils.getInstance().resolveFile(unCompressUri);
			String localfile = source.toString();
			LogWriter.addLog("DEBUG", "本地压缩文件URL：{}", localfile);
			target.copyFrom(source, Selectors.SELECT_ALL);
		}catch(Exception e){
			LogWriter.addLog("ERROR", "文件{}解压写入到{}失败,{}", fileTransInfo.getSourceVfsURI(), fileTransInfo.getTargetVfsURI(),
					e.getMessage());
			return;
		}finally{
			DpVfsUtils.getInstance().closeFileObject(source);
			DpVfsUtils.getInstance().closeFileObject(target);
			if(!fileTransInfo.isKeepBack()){
				DpVfsUtils.getInstance().closeFileObject(sourceFile);
				DpVfsUtils.getInstance().closeFileObject(targetTempFile);
				DpVfsUtils.getInstance().closeFileObject(targetFile);
			}
		}
	}

	private void closeInStream() throws IOException {
		if (bufferReader != null)
			bufferReader.close();
		if (in != null)
			in.close();
		if (reader != null)
			reader.close();
	}
	public void closeOutStream() throws IOException{
		
		if (bufferWriter != null)
			bufferWriter.close();
		if (out != null)
			out.close();
		if (writer != null)
			writer.close();
	}
	public void close(){
		
		try {
			if(targetTempFile.exists()){
				targetTempFile.moveTo(targetFile);
			}
			closeOutStream();
		} catch (IOException e) {
			LogWriter.addLog( "ERROR",e.getMessage());
			e.printStackTrace();
		}
		DpVfsUtils.getInstance().closeFileObject(targetTempFile);
		DpVfsUtils.getInstance().closeFileObject(targetFile);
	}

	private void copyFileByLine() throws IOException, ScriptException {
		bufferReader = new LineReader(reader);
		bufferWriter = new BufferedWriter(writer);
		String lineSeparator = System.getProperty("line.separator", "\n");
		int lineSeparatorLength = lineSeparator.length();
		bufferReader.setLineSeparator(lineSeparator);
		String row = null;
		long rowNum = 1;
		long startRowNum = 1;
		String startRowNumStr = fileTransInfo.getStartRowNum();
		if (StringUtils.isNumberic(startRowNumStr)) {
			startRowNum = Long.valueOf(startRowNumStr);
		}
		
		String fileFullPath=fileTransInfo.getTargetVfsURI();
		while ((row = bufferReader.readLine()) != null && row.length() > 0) {
			if (rowNum >= startRowNum) {
				row = rowProcessor(row);
				int rowLength = row.length();
				bufferWriter.write(row, 0, rowLength);
				bufferWriter.write(lineSeparator, 0, lineSeparatorLength);
				transSize+=rowLength+lineSeparatorLength;
				long fileSize=transSize+sourceSize;
				if(isRelocate(fileSize,rowNum)){
					int index=fileTransInfo.getTargetVfsURI().toString().indexOf(".");
					String filePath=fileTransInfo.getTargetVfsURI().toString();
					
					fileFullPath=getFileName(filePath,index,counter);
					targetFile=DpVfsUtils.getInstance().resolveFile(
							fileFullPath);
					targetTempFile.moveTo(targetFile);
					LogWriter.addLog("INFO","文件输出到{} 文件大小：{}M,行数:{}",fileFullPath,fileSize/1024.0/1024.0+"",""+rowNum);
					
					closeOutStream();
					DpVfsUtils.getInstance().closeFileObject(targetTempFile);
					DpVfsUtils.getInstance().closeFileObject(targetFile);
					
					counter++;
					fileTransInfo.setMerged(true);
					fileFullPath=fileTransInfo.getTargetVfsURI().toString();
					setUpTargetObjects(fileFullPath);
					bufferWriter = new BufferedWriter(writer);

					transSize=0;
					sourceSize=0;
					rowNum=0;
				}
				
			}
			rowNum++;
		}
		fileTransInfo.setTargetVfsURI(fileFullPath);
		fileTransInfo.setRowNum(rowNum - startRowNum);
		
	}
	protected boolean isRelocate(long fileSize,long rowNum) throws FileSystemException{
		if(fileTransInfo.getSplitType().equalsIgnoreCase(splitBySize))
		{
			if(fileSize>=fileTransInfo.getSplitSize()*1024*1024)
				return true;
			else
				return false;
		}else if(fileTransInfo.getSplitType().equalsIgnoreCase(splitByRowNum)){
			if(rowNum>=fileTransInfo.getSplitRowNum())
				return true;
			else
				return false;
		}else if(fileTransInfo.isAppend()&&fileSize>=fileTransInfo.getFileMergeSize()){
			return true;
		}else
			return false;
		
	}

	protected String getFileName(String filePath,int index,int counter){
		String fileFullPath="";
		if(index>0)
			fileFullPath=filePath.substring(0, index)+"_"+timestamp+"_0"+counter+filePath.substring(index);
		else
			fileFullPath=filePath+"_"+timestamp+"_0"+counter;
		return fileFullPath;
	}

	private void copyFileBySize() throws IOException {
		transSize = IOUtils.copyLarge(in, out);
		out.flush();
		long fileSize=transSize+sourceSize;
		
		if(isRelocate(fileSize,0)){
			int index=fileTransInfo.getTargetVfsURI().toString().indexOf(".");
			String filePath=fileTransInfo.getTargetVfsURI().toString();
			
			String fileFullPath=getFileName(filePath,index,counter);
			targetFile=DpVfsUtils.getInstance().resolveFile(
					fileFullPath);
			fileTransInfo.setTargetVfsURI(fileFullPath);
			fileTransInfo.setMerged(true);
		}
	}

	private String rowProcessor(String row) throws ScriptException {
		if (fileTransInfo.getRowProcessScript() != null
				&& fileTransInfo.getRowProcessScript().length() > 0) {

			if (compiledScript == null) {
				ScriptEngineManager sem = new ScriptEngineManager();
				ScriptEngine engine = null;
				try {
					engine = sem.getEngineByName("nashorn");
					if (engine == null) {
						LogWriter.addLog("DEBUG",
								"创建nashorn引擎失败，未使用Jdk 8，使用默认Rhino引擎");
						engine = sem.getEngineByName("JavaScript");
					}
				} catch (Exception e) {
					LogWriter.addLog("DEBUG",
							"创建nashorn引擎失败，不是使用Jdk 8使用默认Rhino引擎，{}",
							e.getMessage());
					engine = sem.getEngineByName("JavaScript");
				}
				Compilable compilable = (Compilable) engine;
				StringBuilder script = new StringBuilder("(function(){ ");
				script.append(fileTransInfo.getRowProcessScript()).append(
						"})()");
				LogWriter.addLog("INFO", "行处理脚本为：{}", script.toString());
				compiledScript = compilable.compile(script.toString());
				bindings = engine.createBindings();
			}
			bindings.put("row", row);
			Object obj = compiledScript.eval(bindings);
			return obj != null ? obj.toString() : "";

		}
		return row;
	}

}
